#ifndef NET_RPC_RPC_CHANNEL_H
#define NET_RPC_RPC_CHANNEL_H

#include <map>
#include <google/protobuf/service.h>
#include <google/protobuf/message.h>
#include "net/base/client_connection.h"
#include "io/byte_stream.h"
#include "base/callback.h"
#include "base/mutex.h"
#include "thread/executor.h"
#include "thread/thread.h"

using std::map;
class Executor;
namespace google {
namespace protobuf {
class Message;
class MethodDescriptor;
class RpcController;
class Rpc;
}
}

struct RpcContext {
  Closure* closure;
  google::protobuf::Message* msg;
  RPC* rpc;
};

class RpcClientConnection : public AsyncClientConnection {
 public:
  RpcClientConnection(const char* ip, int port, Executor* executor);
  virtual void HandleRecv(char* buffer, int size);
  virtual void HandleDisconnect();
  void HandleRecvInternal(char* buffer, int size);
  void SendMessage(RPC* rpc, int64* req_id, const string& name, const google::protobuf::Message& req0,
      google::protobuf::Message* resp, Closure* callback);

 private:
  void DeadlineExceeded(int64 req_id);
  void FinalizeRPC(int64 req_id, char* buffer, int buffer_size);
  void SendInternal(string* s1, string* s2);
  Mutex recv_mutex_;
  Mutex rpc_context_mutex_;
  ByteStream read_buf;
  map<int64, RpcContext>  rpc_context_;
  Executor* executor_;
};

class LoopThread : public Thread {
 public:
  LoopThread(ClientPollRunner* runner) : runner_(runner) {
  }
  virtual void Run() {
    runner_->Loop();
  }
 private:
  ClientPollRunner* runner_;
};

class ClientRpcChannel : public google::protobuf::RpcChannel {
 public:
  ClientRpcChannel(const char* ip, int port, ClientPollRunner* runner, Executor* executor);
  virtual void CallMethod(const google::protobuf::MethodDescriptor* method,
                          google::protobuf::RpcController* controller,
                          const google::protobuf::Message* request,
                          google::protobuf::Message* response,
                          Closure* done);
 private:
  RpcClientConnection conn_;
  int64 i_;
};
#endif
